package demo.main;

import demo.kafka.streams.utils.Constants;
import demo.kafka.streams.utils.KafkaUtil;
import demo.kafka.streams.utils.ThreadUtil;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * 只读应用
 * <ol>
 *     <li>生产者发送文本行到 {@value #TOPIC_TEXT_LINES} 主题</li>
 *     <li>应用读到消息后打印</li>
 * </ol>
 */
public class ReadOnlyApp {

    private static final Logger logger = LoggerFactory.getLogger(ReadOnlyApp.class);

    public static final String TOPIC_TEXT_LINES = "test.stan";
    private static KafkaStreams kafkaStreams;

    public static void main(final String[] args) throws Exception {
        KafkaUtil.startProducer(TOPIC_TEXT_LINES, true);

        Properties props = new Properties();
        // APPLICATION_ID_CONFIG 实际上就是消费组ID
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "read-only-app");
        // 从最早的消息开始消费
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, Constants.KAFKA_BOOTSTRAP_SERVERS);
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000); // 1秒提交一次

        StreamsBuilder streamsBuilder = new StreamsBuilder();
        KStream<String, String> textLines = streamsBuilder.stream(TOPIC_TEXT_LINES);
        textLines.foreach((key, value) -> logger.info("收到消息: key={}, value={}", key, value));

        Topology topology = streamsBuilder.build();
        kafkaStreams = new KafkaStreams(topology, props);
        kafkaStreams.start();

        System.err.println("启动成功，应用ID: " + props.getProperty(StreamsConfig.APPLICATION_ID_CONFIG));


        System.err.println("Kafka Streams 状态: " + kafkaStreams.state());
        // 等待Streams完全启动再查询
        while (kafkaStreams.state() != KafkaStreams.State.RUNNING) {
            ThreadUtil.safeSleep(500);
        }
        System.err.println("Kafka Streams 状态: " + kafkaStreams.state());
        ThreadUtil.safeSleep(30000);
        kafkaStreams.close();

        while (kafkaStreams.state() != KafkaStreams.State.NOT_RUNNING) {
            Thread.sleep(1000);
            System.out.println("等待关闭完成，当前状态: " + kafkaStreams.state());
        }
        // 添加关闭钩子，优雅关闭
//        Runtime.getRuntime().addShutdownHook(new Thread(kafkaStreams::close));

    }
}
